use log::{error, info, warn};
use rdkafka::consumer::{CommitMode, Consumer};
use rdkafka::message::Headers;
use rdkafka::Message;
use subprocess::Exec;

use crate::consumer::make_consumer;
use crate::structs::kafka::Message as KafkaMessage;

fn process_message(msg: KafkaMessage) {
    info!("{}", msg.msg_to_send());
    let cv = msg.msg_to_send();
    let cmd_line = format!(
        "bash -c 'echo -e {} | sudo pocsag -f {} -b 0 -t 1'",
        cv, msg.freq
    );
    info!("{}", cmd_line);
    let f = Exec::shell(cmd_line).join();
    info!("{:?}", f);
}

pub(crate) async fn consume_and_send(
    brokers: &str,
    group_id: &str,
    topics: &[&str],
    username: Option<&str>,
    password: &str,
) {
    let consumer = make_consumer(brokers, group_id, username, password);
    consumer
        .subscribe(topics)
        .expect("Can't subscribe to specified topics");
    loop {
        match consumer.recv().await {
            Err(e) => warn!("Kafka error: {}", e),
            Ok(m) => {
                let payload = match m.payload_view::<str>() {
                    None => "",
                    Some(Ok(s)) => s,
                    Some(Err(e)) => {
                        error!("Error while deserializing message payload: {:?}", e);
                        ""
                    }
                };
                info!("key: '{:?}', payload: '{}', topic: {}, partition: {}, offset: {}, timestamp: {:?}",
                      m.key(), payload, m.topic(), m.partition(), m.offset(), m.timestamp());
                if let Some(headers) = m.headers() {
                    for i in 0..headers.count() {
                        let header = headers.get(i);
                        info!("  Header {:#?}: {:?}", header.key, header.value);
                    }
                }
                match serde_json::from_str(payload) {
                    Ok(v) => process_message(v),
                    Err(e) => error!("Not a message. Err: {}", e),
                };
                consumer.commit_message(&m, CommitMode::Async).unwrap();
            }
        };
    }
}
